package cn.wapache.timed.spi;

import java.net.URI;
import java.util.HashMap;
import java.util.Map;

import cn.wapache.time.calendar.Calendar;
import cn.wapache.timed.Scheduler;
import cn.wapache.timed.serialization.JdkSerializer;
import cn.wapache.timed.serialization.Serializer;
import cn.wapache.timed.zookeeper.ZooKeeperScheduler;

public class SchedulerProvider implements AutoCloseable {

	Map<String, Scheduler> schedulers = new HashMap<>();
	
	public Scheduler newScheduler(URI uri){
		Scheduler scheduler = null;
		switch(uri.getScheme()){
		case "zookeeper":
			scheduler = new ZooKeeperScheduler(this, ZooKeeperScheduler.Options.from(uri));
			break;
		default:
			throw new IllegalArgumentException("unknown scheme :"+uri.getScheme());
		}
		
		synchronized (schedulers) {
			if(schedulers.containsKey(scheduler.getId())){
				throw new IllegalArgumentException("scheduler already exists :"+scheduler.getId());
			}else{
				schedulers.put(scheduler.getId(), scheduler);
			}
		}
		
		return scheduler;
	}

	@SuppressWarnings("unchecked")
	public <S extends Scheduler> S getScheduler(String id){
		synchronized (schedulers) {
			Scheduler scheduler = schedulers.get(id);
			if(scheduler==null){
				throw new IllegalArgumentException("scheduler not exists :"+id);
			}
			return (S)scheduler;
		}
	}

	public void removeScheduler(String id) {
		synchronized (schedulers) {
			Scheduler scheduler = schedulers.remove(id);
			if(scheduler!=null){
				scheduler.stop();
			}
		}
	}
	
	public Serializer getSerializer(String id){
		switch(id.toLowerCase()){
		case "jdk":
		case "java":
		case "default":
			return new JdkSerializer();
		default:
			throw new IllegalArgumentException("serializer not exists :"+id);
		}
	}
	
	public void close(){
		synchronized (schedulers) {
			schedulers.values().forEach(s->s.stop());
			schedulers.clear();
		}
	}

	
}
